package a_o2odata_deal.utils

import java.text.SimpleDateFormat
import java.util.Date

import com.alibaba.fastjson.{JSON, JSONArray}
import org.apache.spark.sql.{DataFrame, SQLContext}

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2019/12/14 10:25
  * @ Param:  ${PARAM}
  * @ Description: 
  */
object handle_zhibo_v2 {

  def handle_zhibo(sqlContext:SQLContext):DataFrame={
    sqlContext.udf.register("tranTimeToString",tranTimeToString _)
    sqlContext.udf.register("handle_array",handle_array _)

   /* val zhibo_data = sqlContext.read.json("s3a://o2o-dataproces-group/panzonghao/zhibo/backups/tmall/11")
      .drop("viewcount","roomTypeName","roomType","nick","liveTitle","liveTime","liveId","is_selfLive",
        "is_showLive","anchorId","commodityCount","liveCount","anchorFans","live_priceText","liveAvgPriceText")*/

    //val zhibo_data = sqlContext.read.orc(s"${source_result}_modify_xiajia")
    val zhibo_data = sqlContext.read.orc("s3a://dws-data/g_data/2020/1/taobao")

    zhibo_data.createOrReplaceTempView("zhibo_data")
    //加载直播数据
    //关联直播
    val data_detail = sqlContext.read.json(s"${a_aa_amainpackage.a_o2odata_deal.config.config.tmall_live_detail_path}")
      .dropDuplicates("good_id","liveId")
    val data_id = sqlContext.read.json(s"${a_aa_amainpackage.a_o2odata_deal.config.config.tmall_live_id_path}")
      //.where("cast(liveTime as long)<1575129600 and cast(liveTime as long)>=1572537600")

    data_id.createOrReplaceTempView("data_id")
    data_detail.createOrReplaceTempView("data_detail")

    //店铺主播当期所有直播场次观看人数的累加，数据处理新增字段：“店铺直播观看人数”
   /* sqlContext.sql(
      s"""
         |select
         |*,
         |sum(viewerCount ) over(partition by anchorId) as shopLive_viewCount
         |from data_id
       """.stripMargin).createOrReplaceTempView("data_id")*/

    //首先统计出单品在多少个直播间出现过,1对多的关系，关联出来数据会增多
    val source_data = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |t2.livePriceText,
         |t2.good_id as good_ids
         |from data_id t1
         |inner join
         |data_detail t2
         |on t1.liveId=t2.liveId
       """.stripMargin).drop("good_id").withColumnRenamed("good_ids","good_id").where("good_id!='-1'")

    source_data.createOrReplaceTempView("source_data")

    sqlContext.sql(
      s"""
         |select
         |good_id,
         |anchorId,
         |livePriceText,
         |roomType,
         |liveId,
         |liveTime,
         |roomTypeName,
         |nick,
         |liveTitle,
         |commodityCount,
         |cast(anchorFans as bigint) as anchorFans,
         |--直播次数
         |count(liveId) over(partition by good_id) as liveCount,
         |--观看人数,基于商品的
         |sum(cast(viewerCount as bigint)) over(partition by good_id) as viewcount,
         |--直播均价
         |avg(cast(livePriceText as double)) over(partition by good_id) as liveAvgPriceText
         |from source_data
       """.stripMargin).dropDuplicates("good_id").createOrReplaceTempView("value")


//每个商品下钻分析（新增）：每条商品每次直播的主播名称、观看人数、直播时间
    sqlContext.sql(
      s"""
         |select distinct
         |good_id,
         |collect_list(liveId) as liveId_collect,
         |collect_list(anchorId) as anchorId_collect,
         |collect_list(nick) as nick_collect,
         |collect_list(viewerCount) as viewerCount_collect,
         |collect_list(tranTimeToString(liveTime)) as liveTime_collect
         |from source_data
         |group by good_id
       """.stripMargin).createOrReplaceTempView("tmp")

    sqlContext.sql(
      s"""
         |select
         |good_id,
         |handle_array(liveId_collect,anchorId_collect,nick_collect,viewerCount_collect,liveTime_collect) as add_to_field
         |from tmp
       """.stripMargin).createOrReplaceTempView("mid")

    //打上add_to_field标签/*
    sqlContext.sql(
      s"""
         |select
         |t1.*,
         |t2.add_to_field
         |from value t1
         |inner join
         |mid t2
         |on t1.good_id=t2.good_id
       """.stripMargin).createOrReplaceTempView("value1")


    //打上是否是自主直播标签
    sqlContext.sql(
      s"""
         |select
         |t1.*,
         |case
         |when t2.userId is null then "false"
         |else "true"
         |end as is_selfLive
         |from value1 t1
         |left join
         |zhibo_data t2
         |on t1.anchorId=t2.userId
       """.stripMargin).dropDuplicates("good_id").createOrReplaceTempView("data")


    val result = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |--t2.anchorId,
         |--t2.livePriceText,
         |--t2.roomType,
         |--t2.liveId,
         |--t2.liveTime,
         |--t2.roomTypeName,
         |--t2.nick,
         |--t2.liveTitle,
         |--t2.commodityCount,
         |--cast(t2.anchorFans as bigint) as anchorFans,
         |case when t2.good_id is null then "" else cast(t2.liveCount as bigint) end as liveCount,
         |case when t2.good_id is null then "" else cast(t2.viewcount as bigint) end as viewcount,
         |case when t2.good_id is null then "" else cast(t2.liveAvgPriceText as double) end as liveAvgPriceText,
         |--t2.add_to_field,
         |case when t2.good_id is null then "" else t2.is_selfLive end as is_selfLive,
         |case when t2.good_id is null then "false" else "true"  end as is_showLive
         |from zhibo_data t1
         |left join
         |data t2
         |on t1.good_id=t2.good_id
       """.stripMargin)
    result.printSchema()
    import org.apache.spark.sql.functions._
    result.agg(count("*"),sum("sellCount"),sum("salesAmount")).show(false)
    result.withColumnRenamed("liveAvgPriceText","liveAvg_priceText")
      //.repartition(12).write.orc("s3a://o2o-dataproces-group/panzonghao/tmall/2020/1/zhibo_finally")
      .repartition(12).write.orc("s3a://o2o-dataproces-group/panzonghao/taobao/2020/1/zhibo_finally")
    null
  }


    def handle_array(liveId_collect:Seq[String],anchorId_collect:Seq[String],nick_collect: Seq[String],viewerCount_collect:Seq[String],liveTime_collect:Seq[String]):String={
    val array = new JSONArray();
    for (i <- 0 to anchorId_collect.length-1){
      val jsonStr = "{" +
        "\"liveId\":"+"\""+liveId_collect(i)+"\"," +
        "\"anchorId\":"+"\""+anchorId_collect(i)+"\"," +
        "\"nick\":"+"\""+nick_collect(i)+"\"," +
        "\"viewCount\":"+"\""+viewerCount_collect(i)+"\"," +
        "\"liveTime\":"+"\""+liveTime_collect(i)+"\"}"
      val jsonObject = JSON.parseObject(jsonStr)
      array.add(jsonObject)
    }
    array.toString
  }


  def tranTimeToString(timestamp:String) :String={
    val tm = timestamp+"000"
    val fm = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    val time = fm.format(new Date(tm.toLong))
    time
  }
}
